/*
 * Copyright (C) 2007 The Guava Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.google.common.util.concurrent;

import com.google.common.annotations.Beta;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Supplier;
import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
import com.google.common.collect.Queues;
import com.google.common.util.concurrent.ForwardingListenableFuture.SimpleForwardingListenableFuture;

import java.lang.reflect.InvocationTargetException;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;

/**
 * Factory and utility methods for {@link java.util.concurrent.Executor}, {@link
 * ExecutorService}, and {@link ThreadFactory}.
 *
 * @author Eric Fellheimer
 * @author Kyle Littlefield
 * @author Justin Mahoney
 * @since 3.0
 */
public final class MoreExecutors {
    private MoreExecutors() {}

    /**
     * Converts the given ThreadPoolExecutor into an ExecutorService that exits
     * when the application is complete.  It does so by using daemon threads and
     * adding a shutdown hook to wait for their completion.
     *
     * <p>This is mainly for fixed thread pools.
     * See {@link Executors#newFixedThreadPool(int)}.
     *
     * @param executor the executor to modify to make sure it exits when the
     *        application is finished
     * @param terminationTimeout how long to wait for the executor to
     *        finish before terminating the JVM
     * @param timeUnit unit of time for the time parameter
     * @return an unmodifiable version of the input which will not hang the JVM
     */
    @Beta
    public static ExecutorService getExitingExecutorService(
            ThreadPoolExecutor executor, long terminationTimeout, TimeUnit timeUnit) {
        return new Application()
                .getExitingExecutorService(executor, terminationTimeout, timeUnit);
    }

    /**
     * Converts the given ScheduledThreadPoolExecutor into a
     * ScheduledExecutorService that exits when the application is complete.  It
     * does so by using daemon threads and adding a shutdown hook to wait for
     * their completion.
     *
     * <p>This is mainly for fixed thread pools.
     * See {@link Executors#newScheduledThreadPool(int)}.
     *
     * @param executor the executor to modify to make sure it exits when the
     *        application is finished
     * @param terminationTimeout how long to wait for the executor to
     *        finish before terminating the JVM
     * @param timeUnit unit of time for the time parameter
     * @return an unmodifiable version of the input which will not hang the JVM
     */
    @Beta
    public static ScheduledExecutorService getExitingScheduledExecutorService(
            ScheduledThreadPoolExecutor executor, long terminationTimeout, TimeUnit timeUnit) {
        return new Application()
                .getExitingScheduledExecutorService(executor, terminationTimeout, timeUnit);
    }

    /**
     * Add a shutdown hook to wait for thread completion in the given
     * {@link ExecutorService service}.  This is useful if the given service uses
     * daemon threads, and we want to keep the JVM from exiting immediately on
     * shutdown, instead giving these daemon threads a chance to terminate
     * normally.
     * @param service ExecutorService which uses daemon threads
     * @param terminationTimeout how long to wait for the executor to finish
     *        before terminating the JVM
     * @param timeUnit unit of time for the time parameter
     */
    @Beta
    public static void addDelayedShutdownHook(
            ExecutorService service, long terminationTimeout, TimeUnit timeUnit) {
        new Application()
                .addDelayedShutdownHook(service, terminationTimeout, timeUnit);
    }

    /**
     * Converts the given ThreadPoolExecutor into an ExecutorService that exits
     * when the application is complete.  It does so by using daemon threads and
     * adding a shutdown hook to wait for their completion.
     *
     * <p>This method waits 120 seconds before continuing with JVM termination,
     * even if the executor has not finished its work.
     *
     * <p>This is mainly for fixed thread pools.
     * See {@link Executors#newFixedThreadPool(int)}.
     *
     * @param executor the executor to modify to make sure it exits when the
     *        application is finished
     * @return an unmodifiable version of the input which will not hang the JVM
     */
    @Beta
    public static ExecutorService getExitingExecutorService(ThreadPoolExecutor executor) {
        return new Application().getExitingExecutorService(executor);
    }

    /**
     * Converts the given ThreadPoolExecutor into a ScheduledExecutorService that
     * exits when the application is complete.  It does so by using daemon threads
     * and adding a shutdown hook to wait for their completion.
     *
     * <p>This method waits 120 seconds before continuing with JVM termination,
     * even if the executor has not finished its work.
     *
     * <p>This is mainly for fixed thread pools.
     * See {@link Executors#newScheduledThreadPool(int)}.
     *
     * @param executor the executor to modify to make sure it exits when the
     *        application is finished
     * @return an unmodifiable version of the input which will not hang the JVM
     */
    @Beta
    public static ScheduledExecutorService getExitingScheduledExecutorService(
            ScheduledThreadPoolExecutor executor) {
        return new Application().getExitingScheduledExecutorService(executor);
    }

    /** Represents the current application to register shutdown hooks. */
    @VisibleForTesting static class Application {

        final ExecutorService getExitingExecutorService(
                ThreadPoolExecutor executor, long terminationTimeout, TimeUnit timeUnit) {
            useDaemonThreadFactory(executor);
            ExecutorService service = Executors.unconfigurableExecutorService(executor);
            addDelayedShutdownHook(service, terminationTimeout, timeUnit);
            return service;
        }

        final ScheduledExecutorService getExitingScheduledExecutorService(
                ScheduledThreadPoolExecutor executor, long terminationTimeout, TimeUnit timeUnit) {
            useDaemonThreadFactory(executor);
            ScheduledExecutorService service = Executors.unconfigurableScheduledExecutorService(executor);
            addDelayedShutdownHook(service, terminationTimeout, timeUnit);
            return service;
        }

        final void addDelayedShutdownHook(
                final ExecutorService service, final long terminationTimeout, final TimeUnit timeUnit) {
            checkNotNull(service);
            checkNotNull(timeUnit);
            addShutdownHook(MoreExecutors.newThread("DelayedShutdownHook-for-" + service, new Runnable() {
                @Override
                public void run() {
                    try {
                        // We'd like to log progress and failures that may arise in the
                        // following code, but unfortunately the behavior of logging
                        // is undefined in shutdown hooks.
                        // This is because the logging code installs a shutdown hook of its
                        // own. See Cleaner class inside {@link LogManager}.
                        service.shutdown();
                        service.awaitTermination(terminationTimeout, timeUnit);
                    } catch (InterruptedException ignored) {
                        // We're shutting down anyway, so just ignore.
                    }
                }
            }));
        }

        final ExecutorService getExitingExecutorService(ThreadPoolExecutor executor) {
            return getExitingExecutorService(executor, 120, TimeUnit.SECONDS);
        }

        final ScheduledExecutorService getExitingScheduledExecutorService(
                ScheduledThreadPoolExecutor executor) {
            return getExitingScheduledExecutorService(executor, 120, TimeUnit.SECONDS);
        }

        @VisibleForTesting void addShutdownHook(Thread hook) {
            Runtime.getRuntime().addShutdownHook(hook);
        }
    }

    private static void useDaemonThreadFactory(ThreadPoolExecutor executor) {
        executor.setThreadFactory(new ThreadFactoryBuilder()
                .setDaemon(true)
                .setThreadFactory(executor.getThreadFactory())
                .build());
    }

    /**
     * Creates an executor service that runs each task in the thread
     * that invokes {@code execute/submit}, as in {@link CallerRunsPolicy}  This
     * applies both to individually submitted tasks and to collections of tasks
     * submitted via {@code invokeAll} or {@code invokeAny}.  In the latter case,
     * tasks will run serially on the calling thread.  Tasks are run to
     * completion before a {@code Future} is returned to the caller (unless the
     * executor has been shutdown).
     *
     * <p>Although all tasks are immediately executed in the thread that
     * submitted the task, this {@code ExecutorService} imposes a small
     * locking overhead on each task submission in order to implement shutdown
     * and termination behavior.
     *
     * <p>The implementation deviates from the {@code ExecutorService}
     * specification with regards to the {@code shutdownNow} method.  First,
     * "best-effort" with regards to canceling running tasks is implemented
     * as "no-effort".  No interrupts or other attempts are made to stop
     * threads executing tasks.  Second, the returned list will always be empty,
     * as any submitted task is considered to have started execution.
     * This applies also to tasks given to {@code invokeAll} or {@code invokeAny}
     * which are pending serial execution, even the subset of the tasks that
     * have not yet started execution.  It is unclear from the
     * {@code ExecutorService} specification if these should be included, and
     * it's much easier to implement the interpretation that they not be.
     * Finally, a call to {@code shutdown} or {@code shutdownNow} may result
     * in concurrent calls to {@code invokeAll/invokeAny} throwing
     * RejectedExecutionException, although a subset of the tasks may already
     * have been executed.
     *
     * @since 10.0 (<a href="http://code.google.com/p/guava-libraries/wiki/Compatibility"
     *        >mostly source-compatible</a> since 3.0)
     * @deprecated Use {@link #directExecutor()} if you only require an {@link Executor} and
     *     {@link #newDirectExecutorService()} if you need a {@link ListeningExecutorService}.
     */
    @Deprecated public static ListeningExecutorService sameThreadExecutor() {
        return new DirectExecutorService();
    }

    // See sameThreadExecutor javadoc for behavioral notes.
    private static class DirectExecutorService
            extends AbstractListeningExecutorService {
        /**
         * Lock used whenever accessing the state variables
         * (runningTasks, shutdown, terminationCondition) of the executor
         */
        private final Lock lock = new ReentrantLock();

        /** Signaled after the executor is shutdown and running tasks are done */
        private final Condition termination = lock.newCondition();

        /*
         * Conceptually, these two variables describe the executor being in
         * one of three states:
         *   - Active: shutdown == false
         *   - Shutdown: runningTasks > 0 and shutdown == true
         *   - Terminated: runningTasks == 0 and shutdown == true
         */
        private int runningTasks = 0;
        private boolean shutdown = false;

        @Override
        public void execute(Runnable command) {
            startTask();
            try {
                command.run();
            } finally {
                endTask();
            }
        }

        @Override
        public boolean isShutdown() {
            lock.lock();
            try {
                return shutdown;
            } finally {
                lock.unlock();
            }
        }

        @Override
        public void shutdown() {
            lock.lock();
            try {
                shutdown = true;
            } finally {
                lock.unlock();
            }
        }

        // See sameThreadExecutor javadoc for unusual behavior of this method.
        @Override
        public List<Runnable> shutdownNow() {
            shutdown();
            return Collections.emptyList();
        }

        @Override
        public boolean isTerminated() {
            lock.lock();
            try {
                return shutdown && runningTasks == 0;
            } finally {
                lock.unlock();
            }
        }

        @Override
        public boolean awaitTermination(long timeout, TimeUnit unit)
                throws InterruptedException {
            long nanos = unit.toNanos(timeout);
            lock.lock();
            try {
                for (;;) {
                    if (isTerminated()) {
                        return true;
                    } else if (nanos <= 0) {
                        return false;
                    } else {
                        nanos = termination.awaitNanos(nanos);
                    }
                }
            } finally {
                lock.unlock();
            }
        }

        /**
         * Checks if the executor has been shut down and increments the running
         * task count.
         *
         * @throws RejectedExecutionException if the executor has been previously
         *         shutdown
         */
        private void startTask() {
            lock.lock();
            try {
                if (isShutdown()) {
                    throw new RejectedExecutionException("Executor already shutdown");
                }
                runningTasks++;
            } finally {
                lock.unlock();
            }
        }

        /**
         * Decrements the running task count.
         */
        private void endTask() {
            lock.lock();
            try {
                runningTasks--;
                if (isTerminated()) {
                    termination.signalAll();
                }
            } finally {
                lock.unlock();
            }
        }
    }

    /**
     * Creates an executor service that runs each task in the thread
     * that invokes {@code execute/submit}, as in {@link CallerRunsPolicy}  This
     * applies both to individually submitted tasks and to collections of tasks
     * submitted via {@code invokeAll} or {@code invokeAny}.  In the latter case,
     * tasks will run serially on the calling thread.  Tasks are run to
     * completion before a {@code Future} is returned to the caller (unless the
     * executor has been shutdown).
     *
     * <p>Although all tasks are immediately executed in the thread that
     * submitted the task, this {@code ExecutorService} imposes a small
     * locking overhead on each task submission in order to implement shutdown
     * and termination behavior.
     *
     * <p>The implementation deviates from the {@code ExecutorService}
     * specification with regards to the {@code shutdownNow} method.  First,
     * "best-effort" with regards to canceling running tasks is implemented
     * as "no-effort".  No interrupts or other attempts are made to stop
     * threads executing tasks.  Second, the returned list will always be empty,
     * as any submitted task is considered to have started execution.
     * This applies also to tasks given to {@code invokeAll} or {@code invokeAny}
     * which are pending serial execution, even the subset of the tasks that
     * have not yet started execution.  It is unclear from the
     * {@code ExecutorService} specification if these should be included, and
     * it's much easier to implement the interpretation that they not be.
     * Finally, a call to {@code shutdown} or {@code shutdownNow} may result
     * in concurrent calls to {@code invokeAll/invokeAny} throwing
     * RejectedExecutionException, although a subset of the tasks may already
     * have been executed.
     *
     * @since 18.0 (present as MoreExecutors.sameThreadExecutor() since 10.0)
     */
    public static ListeningExecutorService newDirectExecutorService() {
        return new DirectExecutorService();
    }

    /**
     * Returns an {@link Executor} that runs each task in the thread that invokes
     * {@link Executor#execute execute}, as in {@link CallerRunsPolicy}.
     *
     * <p>This instance is equivalent to: <pre>   {@code
     *   final class DirectExecutor implements Executor {
     *     public void execute(Runnable r) {
     *       r.run();
     *     }
     *   }}</pre>
     *
     * <p>This should be preferred to {@link #newDirectExecutorService()} because the implementing the
     * {@link ExecutorService} subinterface necessitates significant performance overhead.
     *
     * @since 18.0
     */
    public static Executor directExecutor() {
        return DirectExecutor.INSTANCE;
    }

    /** See {@link #directExecutor} for behavioral notes. */
    private enum DirectExecutor implements Executor {
        INSTANCE;
        @Override public void execute(Runnable command) {
            command.run();
        }
    }

    /**
     * Creates an {@link ExecutorService} whose {@code submit} and {@code
     * invokeAll} methods submit {@link ListenableFutureTask} instances to the
     * given delegate executor. Those methods, as well as {@code execute} and
     * {@code invokeAny}, are implemented in terms of calls to {@code
     * delegate.execute}. All other methods are forwarded unchanged to the
     * delegate. This implies that the returned {@code ListeningExecutorService}
     * never calls the delegate's {@code submit}, {@code invokeAll}, and {@code
     * invokeAny} methods, so any special handling of tasks must be implemented in
     * the delegate's {@code execute} method or by wrapping the returned {@code
     * ListeningExecutorService}.
     *
     * <p>If the delegate executor was already an instance of {@code
     * ListeningExecutorService}, it is returned untouched, and the rest of this
     * documentation does not apply.
     *
     * @since 10.0
     */
    public static ListeningExecutorService listeningDecorator(
            ExecutorService delegate) {
        return (delegate instanceof ListeningExecutorService)
                ? (ListeningExecutorService) delegate
                : (delegate instanceof ScheduledExecutorService)
                ? new ScheduledListeningDecorator((ScheduledExecutorService) delegate)
                : new ListeningDecorator(delegate);
    }

    /**
     * Creates a {@link ScheduledExecutorService} whose {@code submit} and {@code
     * invokeAll} methods submit {@link ListenableFutureTask} instances to the
     * given delegate executor. Those methods, as well as {@code execute} and
     * {@code invokeAny}, are implemented in terms of calls to {@code
     * delegate.execute}. All other methods are forwarded unchanged to the
     * delegate. This implies that the returned {@code
     * ListeningScheduledExecutorService} never calls the delegate's {@code
     * submit}, {@code invokeAll}, and {@code invokeAny} methods, so any special
     * handling of tasks must be implemented in the delegate's {@code execute}
     * method or by wrapping the returned {@code
     * ListeningScheduledExecutorService}.
     *
     * <p>If the delegate executor was already an instance of {@code
     * ListeningScheduledExecutorService}, it is returned untouched, and the rest
     * of this documentation does not apply.
     *
     * @since 10.0
     */
    public static ListeningScheduledExecutorService listeningDecorator(
            ScheduledExecutorService delegate) {
        return (delegate instanceof ListeningScheduledExecutorService)
                ? (ListeningScheduledExecutorService) delegate
                : new ScheduledListeningDecorator(delegate);
    }

    private static class ListeningDecorator
            extends AbstractListeningExecutorService {
        private final ExecutorService delegate;

        ListeningDecorator(ExecutorService delegate) {
            this.delegate = checkNotNull(delegate);
        }

        @Override
        public boolean awaitTermination(long timeout, TimeUnit unit)
                throws InterruptedException {
            return delegate.awaitTermination(timeout, unit);
        }

        @Override
        public boolean isShutdown() {
            return delegate.isShutdown();
        }

        @Override
        public boolean isTerminated() {
            return delegate.isTerminated();
        }

        @Override
        public void shutdown() {
            delegate.shutdown();
        }

        @Override
        public List<Runnable> shutdownNow() {
            return delegate.shutdownNow();
        }

        @Override
        public void execute(Runnable command) {
            delegate.execute(command);
        }
    }

    private static class ScheduledListeningDecorator
            extends ListeningDecorator implements ListeningScheduledExecutorService {
        @SuppressWarnings("hiding")
        final ScheduledExecutorService delegate;

        ScheduledListeningDecorator(ScheduledExecutorService delegate) {
            super(delegate);
            this.delegate = checkNotNull(delegate);
        }

        @Override
        public ListenableScheduledFuture<?> schedule(
                Runnable command, long delay, TimeUnit unit) {
            ListenableFutureTask<Void> task =
                    ListenableFutureTask.create(command, null);
            ScheduledFuture<?> scheduled = delegate.schedule(task, delay, unit);
            return new ListenableScheduledTask<Void>(task, scheduled);
        }

        @Override
        public <V> ListenableScheduledFuture<V> schedule(
                Callable<V> callable, long delay, TimeUnit unit) {
            ListenableFutureTask<V> task = ListenableFutureTask.create(callable);
            ScheduledFuture<?> scheduled = delegate.schedule(task, delay, unit);
            return new ListenableScheduledTask<V>(task, scheduled);
        }

        @Override
        public ListenableScheduledFuture<?> scheduleAtFixedRate(
                Runnable command, long initialDelay, long period, TimeUnit unit) {
            NeverSuccessfulListenableFutureTask task =
                    new NeverSuccessfulListenableFutureTask(command);
            ScheduledFuture<?> scheduled =
                    delegate.scheduleAtFixedRate(task, initialDelay, period, unit);
            return new ListenableScheduledTask<Void>(task, scheduled);
        }

        @Override
        public ListenableScheduledFuture<?> scheduleWithFixedDelay(
                Runnable command, long initialDelay, long delay, TimeUnit unit) {
            NeverSuccessfulListenableFutureTask task =
                    new NeverSuccessfulListenableFutureTask(command);
            ScheduledFuture<?> scheduled =
                    delegate.scheduleWithFixedDelay(task, initialDelay, delay, unit);
            return new ListenableScheduledTask<Void>(task, scheduled);
        }

        private static final class ListenableScheduledTask<V>
                extends SimpleForwardingListenableFuture<V>
                implements ListenableScheduledFuture<V> {

            private final ScheduledFuture<?> scheduledDelegate;

            public ListenableScheduledTask(
                    ListenableFuture<V> listenableDelegate,
                    ScheduledFuture<?> scheduledDelegate) {
                super(listenableDelegate);
                this.scheduledDelegate = scheduledDelegate;
            }

            @Override
            public boolean cancel(boolean mayInterruptIfRunning) {
                boolean cancelled = super.cancel(mayInterruptIfRunning);
                if (cancelled) {
                    // Unless it is cancelled, the delegate may continue being scheduled
                    scheduledDelegate.cancel(mayInterruptIfRunning);

                    // TODO(user): Cancel "this" if "scheduledDelegate" is cancelled.
                }
                return cancelled;
            }

            @Override
            public long getDelay(TimeUnit unit) {
                return scheduledDelegate.getDelay(unit);
            }

            @Override
            public int compareTo(Delayed other) {
                return scheduledDelegate.compareTo(other);
            }
        }

        private static final class NeverSuccessfulListenableFutureTask
                extends AbstractFuture<Void>
                implements Runnable {
            private final Runnable delegate;

            public NeverSuccessfulListenableFutureTask(Runnable delegate) {
                this.delegate = checkNotNull(delegate);
            }

            @Override public void run() {
                try {
                    delegate.run();
                } catch (Throwable t) {
                    setException(t);
                    throw Throwables.propagate(t);
                }
            }
        }
    }

  /*
   * This following method is a modified version of one found in
   * http://gee.cs.oswego.edu/cgi-bin/viewcvs.cgi/jsr166/src/test/tck/AbstractExecutorServiceTest.java?revision=1.30
   * which contained the following notice:
   *
   * Written by Doug Lea with assistance from members of JCP JSR-166
   * Expert Group and released to the public domain, as explained at
   * http://creativecommons.org/publicdomain/zero/1.0/
   * Other contributors include Andrew Wright, Jeffrey Hayes,
   * Pat Fisher, Mike Judd.
   */

    /**
     * An implementation of {@link ExecutorService#invokeAny} for {@link ListeningExecutorService}
     * implementations.
     */ static <T> T invokeAnyImpl(ListeningExecutorService executorService,
                                   Collection<? extends Callable<T>> tasks, boolean timed, long nanos)
            throws InterruptedException, ExecutionException, TimeoutException {
        checkNotNull(executorService);
        int ntasks = tasks.size();
        checkArgument(ntasks > 0);
        List<Future<T>> futures = Lists.newArrayListWithCapacity(ntasks);
        BlockingQueue<Future<T>> futureQueue = Queues.newLinkedBlockingQueue();

        // For efficiency, especially in executors with limited
        // parallelism, check to see if previously submitted tasks are
        // done before submitting more of them. This interleaving
        // plus the exception mechanics account for messiness of main
        // loop.

        try {
            // Record exceptions so that if we fail to obtain any
            // result, we can throw the last exception we got.
            ExecutionException ee = null;
            long lastTime = timed ? System.nanoTime() : 0;
            Iterator<? extends Callable<T>> it = tasks.iterator();

            futures.add(submitAndAddQueueListener(executorService, it.next(), futureQueue));
            --ntasks;
            int active = 1;

            for (;;) {
                Future<T> f = futureQueue.poll();
                if (f == null) {
                    if (ntasks > 0) {
                        --ntasks;
                        futures.add(submitAndAddQueueListener(executorService, it.next(), futureQueue));
                        ++active;
                    } else if (active == 0) {
                        break;
                    } else if (timed) {
                        f = futureQueue.poll(nanos, TimeUnit.NANOSECONDS);
                        if (f == null) {
                            throw new TimeoutException();
                        }
                        long now = System.nanoTime();
                        nanos -= now - lastTime;
                        lastTime = now;
                    } else {
                        f = futureQueue.take();
                    }
                }
                if (f != null) {
                    --active;
                    try {
                        return f.get();
                    } catch (ExecutionException eex) {
                        ee = eex;
                    } catch (RuntimeException rex) {
                        ee = new ExecutionException(rex);
                    }
                }
            }

            if (ee == null) {
                ee = new ExecutionException(null);
            }
            throw ee;
        } finally {
            for (Future<T> f : futures) {
                f.cancel(true);
            }
        }
    }

    /**
     * Submits the task and adds a listener that adds the future to {@code queue} when it completes.
     */
    private static <T> ListenableFuture<T> submitAndAddQueueListener(
            ListeningExecutorService executorService, Callable<T> task,
            final BlockingQueue<Future<T>> queue) {
        final ListenableFuture<T> future = executorService.submit(task);
        future.addListener(new Runnable() {
            @Override public void run() {
                queue.add(future);
            }
        }, directExecutor());
        return future;
    }

    /**
     * Returns a default thread factory used to create new threads.
     *
     * <p>On AppEngine, returns {@code ThreadManager.currentRequestThreadFactory()}.
     * Otherwise, returns {@link Executors#defaultThreadFactory()}.
     *
     * @since 14.0
     */
    @Beta
    public static ThreadFactory platformThreadFactory() {
        if (!isAppEngine()) {
            return Executors.defaultThreadFactory();
        }
        try {
            return (ThreadFactory) Class.forName("com.google.appengine.api.ThreadManager")
                    .getMethod("currentRequestThreadFactory")
                    .invoke(null);
        } catch (IllegalAccessException e) {
            throw new RuntimeException("Couldn't invoke ThreadManager.currentRequestThreadFactory", e);
        } catch (ClassNotFoundException e) {
            throw new RuntimeException("Couldn't invoke ThreadManager.currentRequestThreadFactory", e);
        } catch (NoSuchMethodException e) {
            throw new RuntimeException("Couldn't invoke ThreadManager.currentRequestThreadFactory", e);
        } catch (InvocationTargetException e) {
            throw Throwables.propagate(e.getCause());
        }
    }

    private static boolean isAppEngine() {
        if (System.getProperty("com.google.appengine.runtime.environment") == null) {
            return false;
        }
        try {
            // If the current environment is null, we're not inside AppEngine.
            return Class.forName("com.google.apphosting.api.ApiProxy")
                    .getMethod("getCurrentEnvironment")
                    .invoke(null) != null;
        } catch (ClassNotFoundException e) {
            // If ApiProxy doesn't exist, we're not on AppEngine at all.
            return false;
        } catch (InvocationTargetException e) {
            // If ApiProxy throws an exception, we're not in a proper AppEngine environment.
            return false;
        } catch (IllegalAccessException e) {
            // If the method isn't accessible, we're not on a supported version of AppEngine;
            return false;
        } catch (NoSuchMethodException e) {
            // If the method doesn't exist, we're not on a supported version of AppEngine;
            return false;
        }
    }

    /**
     * Creates a thread using {@link #platformThreadFactory}, and sets its name to {@code name}
     * unless changing the name is forbidden by the security manager.
     */
    static Thread newThread(String name, Runnable runnable) {
        checkNotNull(name);
        checkNotNull(runnable);
        Thread result = platformThreadFactory().newThread(runnable);
        try {
            result.setName(name);
        } catch (SecurityException e) {
            // OK if we can't set the name in this environment.
        }
        return result;
    }

    // TODO(user): provide overloads for ListeningExecutorService? ListeningScheduledExecutorService?
    // TODO(user): provide overloads that take constant strings? Function<Runnable, String>s to
    // calculate names?

    /**
     * Creates an {@link Executor} that renames the {@link Thread threads} that its tasks run in.
     *
     * <p>The names are retrieved from the {@code nameSupplier} on the thread that is being renamed
     * right before each task is run.  The renaming is best effort, if a {@link SecurityManager}
     * prevents the renaming then it will be skipped but the tasks will still execute.
     *
     *
     * @param executor The executor to decorate
     * @param nameSupplier The source of names for each task
     */
    static Executor renamingDecorator(final Executor executor, final Supplier<String> nameSupplier) {
        checkNotNull(executor);
        checkNotNull(nameSupplier);
        if (isAppEngine()) {
            // AppEngine doesn't support thread renaming, so don't even try
            return executor;
        }
        return new Executor() {
            @Override public void execute(Runnable command) {
                executor.execute(Callables.threadRenaming(command, nameSupplier));
            }
        };
    }

    /**
     * Creates an {@link ExecutorService} that renames the {@link Thread threads} that its tasks run
     * in.
     *
     * <p>The names are retrieved from the {@code nameSupplier} on the thread that is being renamed
     * right before each task is run.  The renaming is best effort, if a {@link SecurityManager}
     * prevents the renaming then it will be skipped but the tasks will still execute.
     *
     *
     * @param service The executor to decorate
     * @param nameSupplier The source of names for each task
     */
    static ExecutorService renamingDecorator(final ExecutorService service,
                                             final Supplier<String> nameSupplier) {
        checkNotNull(service);
        checkNotNull(nameSupplier);
        if (isAppEngine()) {
            // AppEngine doesn't support thread renaming, so don't even try.
            return service;
        }
        return new WrappingExecutorService(service) {
            @Override protected <T> Callable<T> wrapTask(Callable<T> callable) {
                return Callables.threadRenaming(callable, nameSupplier);
            }
            @Override protected Runnable wrapTask(Runnable command) {
                return Callables.threadRenaming(command, nameSupplier);
            }
        };
    }

    /**
     * Creates a {@link ScheduledExecutorService} that renames the {@link Thread threads} that its
     * tasks run in.
     *
     * <p>The names are retrieved from the {@code nameSupplier} on the thread that is being renamed
     * right before each task is run.  The renaming is best effort, if a {@link SecurityManager}
     * prevents the renaming then it will be skipped but the tasks will still execute.
     *
     *
     * @param service The executor to decorate
     * @param nameSupplier The source of names for each task
     */
    static ScheduledExecutorService renamingDecorator(final ScheduledExecutorService service,
                                                      final Supplier<String> nameSupplier) {
        checkNotNull(service);
        checkNotNull(nameSupplier);
        if (isAppEngine()) {
            // AppEngine doesn't support thread renaming, so don't even try.
            return service;
        }
        return new WrappingScheduledExecutorService(service) {
            @Override protected <T> Callable<T> wrapTask(Callable<T> callable) {
                return Callables.threadRenaming(callable, nameSupplier);
            }
            @Override protected Runnable wrapTask(Runnable command) {
                return Callables.threadRenaming(command, nameSupplier);
            }
        };
    }

    /**
     * Shuts down the given executor gradually, first disabling new submissions and later cancelling
     * existing tasks.
     *
     * <p>The method takes the following steps:
     * <ol>
     *  <li>calls {@link ExecutorService#shutdown()}, disabling acceptance of new submitted tasks.
     *  <li>waits for half of the specified timeout.
     *  <li>if the timeout expires, it calls {@link ExecutorService#shutdownNow()}, cancelling
     *  pending tasks and interrupting running tasks.
     *  <li>waits for the other half of the specified timeout.
     * </ol>
     *
     * <p>If, at any step of the process, the given executor is terminated or the calling thread is
     * interrupted, the method calls {@link ExecutorService#shutdownNow()}, cancelling
     * pending tasks and interrupting running tasks.
     *
     * @param service the {@code ExecutorService} to shut down
     * @param timeout the maximum time to wait for the {@code ExecutorService} to terminate
     * @param unit the time unit of the timeout argument
     * @return {@code true} if the pool was terminated successfully, {@code false} if the
     *     {@code ExecutorService} could not terminate <b>or</b> the thread running this method
     *     is interrupted while waiting for the {@code ExecutorService} to terminate
     * @since 17.0
     */
    @Beta
    public static boolean shutdownAndAwaitTermination(
            ExecutorService service, long timeout, TimeUnit unit) {
        checkNotNull(unit);
        // Disable new tasks from being submitted
        service.shutdown();
        try {
            long halfTimeoutNanos = TimeUnit.NANOSECONDS.convert(timeout, unit) / 2;
            // Wait for half the duration of the timeout for existing tasks to terminate
            if (!service.awaitTermination(halfTimeoutNanos, TimeUnit.NANOSECONDS)) {
                // Cancel currently executing tasks
                service.shutdownNow();
                // Wait the other half of the timeout for tasks to respond to being cancelled
                service.awaitTermination(halfTimeoutNanos, TimeUnit.NANOSECONDS);
            }
        } catch (InterruptedException ie) {
            // Preserve interrupt status
            Thread.currentThread().interrupt();
            // (Re-)Cancel if current thread also interrupted
            service.shutdownNow();
        }
        return service.isTerminated();
    }
}